/**
 * 
 */
package mapreduce4j;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.WritableComparable;

/**
 * The base functionality for the context passed around the Job
 * @author tim
 */
public abstract class MapContext<KEY_IN extends WritableComparable, VALUE_IN extends WritableComparable, KEY_OUT extends WritableComparable,VALUE_OUT extends WritableComparable> {
	// might we consider making this static? 
	// if so we would loose the class name, or have to delegate to 
	// subclasses a 'getLogger()' method
	protected Logger logger = Logger.getLogger(this.getClass().getName());
	
	protected RecordReader<KEY_IN,VALUE_IN> reader;
	
	// job config
	protected Configuration config;
	
	
	protected RecordWriter<KEY_OUT,VALUE_OUT> writer;
	
	
	public MapContext(Configuration config, RecordReader<KEY_IN, VALUE_IN> reader,
			RecordWriter<KEY_OUT, VALUE_OUT> writer) {
		this.reader = reader;
		this.config = config;
		this.writer = writer;
	}

	/**
	 * Logs the status
	 * @param status to log
	 */
	public void setStatus(String status) {
		//logger.log(Level.INFO, status);
	}
	
	/**
	 * @return the job configuration
	 */
	public Configuration getConfiguration() {
		return config;
	}
	
	/**
	 * @param key to write to the job output
	 * @param value to write to the output
	 */
	public void write(KEY_OUT key, VALUE_OUT value) {
		try {
			writer.write(key, value);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public KEY_IN getCurrentKey() throws IOException, InterruptedException {
		return reader.getCurrentKey();
	}

	public VALUE_IN getCurrentValue() throws IOException,
			InterruptedException {
		return reader.getCurrentValue();
	}

	public boolean nextKeyValue() throws IOException, InterruptedException {
		return reader.nextKeyValue();
	}
	
	public void close() throws IOException, InterruptedException {
		writer.close();
	}
}
